package jrain.iot.server.mqtt.business;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import jrain.fw.core.business.Business;
import jrain.fw.core.business.BusinessConst;
import jrain.fw.core.business.BusinessMethod;
import jrain.fw.core.business.BusinessRequest;
import jrain.fw.core.business.BusinessResponse;
import jrain.utils.lang.StringUtils;

@Business(name = "MqttSimpleCtxService")
public class MqttSimpleCtxService {

	private static ConcurrentMap<String, ChannelHandlerContext> ctxMap = new ConcurrentHashMap<>();
	private static final AttributeKey<Integer> countKey = AttributeKey.valueOf("countKey");

	@BusinessMethod()
	public BusinessResponse put(BusinessRequest request) {
		BusinessResponse response = new BusinessResponse();
		String key = StringUtils.trimNull(request.getData("key"));
		if (key.length() > 0) {
			ChannelHandlerContext value = (ChannelHandlerContext) request.getData("value");
			value.channel().attr(countKey).set(0);
			ctxMap.put(key, value);
		}
		return response;
	}

	@BusinessMethod()
	public BusinessResponse remove(BusinessRequest request) {
		BusinessResponse response = new BusinessResponse();
		String key = StringUtils.trimNull(request.getData("key"));
		if (key.length() > 0) {
			ctxMap.remove(key);
		}
		return response;
	}

	@BusinessMethod()
	public BusinessResponse publish(BusinessRequest request) {
		BusinessResponse response = new BusinessResponse();
		String key = StringUtils.trimNull(request.getData("key"));
		String value = StringUtils.trimNull(request.getData("value"));
		if (key.length() > 0 && value.length() > 0) {
			ChannelHandlerContext ctx = ctxMap.get(key);
			if (ctx != null) {
				// 线程安全处理
				boolean b = false;
				int newValue = 0;
				do {
					Attribute<Integer> countAttr = ctx.channel().attr(countKey);
					Integer oldValue = countAttr.get();
					newValue = oldValue + 1;
					if (newValue > 65535) {
						newValue = 0;
					}
					b = countAttr.compareAndSet(oldValue, new Integer(newValue));
				} while (!b);
				MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false,
						MqttQoS.AT_MOST_ONCE, false, 0);
				MqttPublishVariableHeader header = new MqttPublishVariableHeader("server", newValue);
				ByteBuf buf = Unpooled.wrappedBuffer(value.getBytes());
				MqttPublishMessage message = new MqttPublishMessage(mqttFixedHeader, header, buf);
				ctx.channel().write(message);
				ctx.channel().flush();
			} else {
				response.setCode(BusinessConst.RES_CODE_ERROR_UNKOWN);
				response.setMsg("设备不存在或者不在线,指令无法下达！");
			}
		} else {
			response.setCode(BusinessConst.RES_CODE_ERROR_UNKOWN);
			response.setMsg("参数和内容不能为空！");
		}
		return response;
	}

}
